package com.fanxuankai.ksql.function.udaf.min;

import com.google.common.collect.Lists;
import io.confluent.ksql.function.udaf.TableUdaf;
import io.confluent.ksql.function.udaf.UdafDescription;
import io.confluent.ksql.function.udaf.UdafFactory;

import java.nio.ByteBuffer;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.List;
import java.util.Objects;

/**
 * 支持 table 的 min 函数
 *
 * @author shenxiaoyang
 */
@UdafDescription(name = "MIN_PRO", description = "最小的", author = "shenxiaoyang")
public class Min {

    private Min() {

    }

    @UdafFactory(description = "Integer 类型")
    public static TableUdaf<Integer, List<Integer>, Integer> forInteger() {
        return new Function<>();
    }

    @UdafFactory(description = "Long 类型")
    public static TableUdaf<Long, List<Long>, Long> forLong() {
        return new Function<>();
    }

    @UdafFactory(description = "Double 类型")
    public static TableUdaf<Double, List<Double>, Double> forDouble() {
        return new Function<>();
    }

    @UdafFactory(description = "String 类型")
    public static TableUdaf<String, List<String>, String> forString() {
        return new Function<>();
    }

    @UdafFactory(description = "Boolean 类型")
    public static TableUdaf<Boolean, List<Boolean>, Boolean> forBoolean() {
        return new Function<>();
    }

    @UdafFactory(description = "Timestamp 类型")
    public static TableUdaf<Timestamp, List<Timestamp>, Timestamp> forTimestamp() {
        return new Function<>();
    }

    @UdafFactory(description = "Time 类型")
    public static TableUdaf<Time, List<Time>, Time> forTime() {
        return new Function<>();
    }

    @UdafFactory(description = "Date 类型")
    public static TableUdaf<Date, List<Date>, Date> forDate() {
        return new Function<>();
    }

    @UdafFactory(description = "ByteBuffer 类型")
    public static TableUdaf<ByteBuffer, List<ByteBuffer>, ByteBuffer> forByteBuffer() {
        return new Function<>();
    }

    private static final class Function<T extends Comparable<? super T>> implements TableUdaf<T, List<T>, T> {
        @Override
        public List<T> initialize() {
            return Lists.newArrayList();
        }

        @Override
        public List<T> aggregate(final T thisValue, final List<T> aggregate) {
            if (!aggregate.contains(thisValue)) {
                aggregate.add(thisValue);
            }
            return aggregate;
        }

        @Override
        public List<T> merge(final List<T> aggOne, final List<T> aggTwo) {
            for (final T thisEntry : aggTwo) {
                if (!aggOne.contains(thisEntry)) {
                    aggOne.add(thisEntry);
                }
            }
            return aggOne;
        }

        @Override
        public T map(final List<T> agg) {
            return agg.stream().filter(Objects::nonNull).min(T::compareTo).orElse(null);
        }

        @Override
        public List<T> undo(T valueToUndo, List<T> aggregateValue) {
            aggregateValue.remove(valueToUndo);
            return aggregateValue;
        }
    }

}
